package com.atguigu.func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.EduConfig;
import com.atguigu.utils.DruidDSUtil;
import com.atguigu.utils.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * ClassName: DimTableProcessFunction
 * Package: com.atguigu.func
 * Description:
 *
 * @Author Lovxy
 * @Create 2023/5/15 14:15
 * @Version 1.0
 */
public class DimTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> stateDescriptor;
    private HashMap<String, TableProcess> tableProcessHashMap;

    public DimTableProcessFunction(MapStateDescriptor<String, TableProcess> stateDescriptor) {
        this.stateDescriptor = stateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        tableProcessHashMap = new HashMap<>();
        //使用JDBC直接读取配置信息表,将数据存入HashMap
        Connection connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/edu_config?" +
                "user=root&password=000000&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        List<TableProcess> tableProcesses = JdbcUtil.queryList(connection,
                "select  * from table_process where sink_type='dim'",
                TableProcess.class, true);
        for (TableProcess tableProcess : tableProcesses) {
            checkTable(tableProcess);
            tableProcessHashMap.put(tableProcess.getSourceTable(),tableProcess);
        }

        connection.close();

    }
    //Value:{"database":"gmall-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,
    // "data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null}}
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {

        //1.获取状态信息
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(stateDescriptor);
        String table = value.getString("table");
        TableProcess tableProcess = broadcastState.get(table);
        TableProcess tableProcess1 = tableProcessHashMap.get(table);
        //type:insert update delete bootstrap-start bootstrap-insert bootstrap-complete
        String type = value.getString("type");
        //2.过滤
        //2.1 行过滤
        if ((tableProcess != null || tableProcess1 != null ) && ("insert".equals(type)||"update".equals(type)||"bootstrap-insert".equals(type))){
            if (tableProcess == null){
                tableProcess=tableProcess1;
            }
            //2.2 列过滤
            filterColumns(value.getJSONObject("data"),tableProcess.getSinkColumns());
            //3.写出
            value.put("sink_table",tableProcess.getSinkTable());
            out.collect(value);
        }else {
            System.out.println("该表:" + table + "没有对应的配置信息！或者类型不匹配:" + type);
        }

    }


    /**
     * 列过滤
     * @param data {"id":"1001","name":"zs","sex":"male","tm_name":"aaa"}
     * @param sinkColumns id , tm_name
     */
    private void filterColumns(JSONObject data, String sinkColumns) {
        //切分sinkColumns 并转化为集合
        String[] split = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(split);
//        Set<Map.Entry<String, Object>> entries = data.entrySet();
//        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
//        while (iterator.hasNext()){
//            Map.Entry<String, Object> next = iterator.next();
//            if (!columnList.contains(next.getKey())){
//                iterator.remove();
//            }
//        }
        //过滤
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !columnList.contains(next.getKey()));

    }
    //value:{"before":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},
    // "after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},
    // "source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1669162876406,"snapshot":"false",
    // "db":"gmall-220623-config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},
    // "op":"r","ts_ms":1669162876406,"transaction":null}

    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(stateDescriptor);


        //0.如果为删除操作 即op为d，则删除状态中的数据
        JSONObject jsonObject = JSONObject.parseObject(value);
        if ("d".equals(jsonObject.getString("op"))) {
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("before"),
                    TableProcess.class);
            String key = tableProcess.getSourceTable();
            broadcastState.remove(key);
            tableProcessHashMap.remove(key);
        } else {
            //1.将数据转换为TableProcess对象
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("after"),
                    TableProcess.class);

            if ("dim".equals(tableProcess.getSinkType())){
                //2.校验并建表
                checkTable(tableProcess);

                //3.将数据写入状态
                broadcastState.put(tableProcess.getSourceTable(),tableProcess);
            }

        }


    }

    private void checkTable(TableProcess tableProcess) {
        //获取连接
        DruidPooledConnection phoenixConn = null;
        PreparedStatement preparedStatement = null;

        //获取建表相关字段
        String sinkTable = tableProcess.getSinkTable();
        String sinkPk = tableProcess.getSinkPk();
        String sinkColumns = tableProcess.getSinkColumns();
        String sinkExtend = tableProcess.getSinkExtend();
        try {
            //封装建表语句create table if not exists db.sinkTable(id varchar primary key ,name varchar,sex varchar)
            phoenixConn = DruidDSUtil.getPhoenixConn();
            if (sinkPk == null || "".equals(sinkPk)) {
                sinkPk = "id";
            }
            if (sinkExtend == null) {
                sinkExtend = "";
            }
            StringBuilder creteTableSQL = new StringBuilder("create table if not exists ")
                    .append(EduConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");

            String[] columnArr = sinkColumns.split(",");
            for (int i = 0; i < columnArr.length; i++) {
                String column = columnArr[i];
                if (sinkPk.equals(column)) {
                    creteTableSQL.append(column).append(" varchar primary key");
                } else {
                    creteTableSQL.append(column).append(" varchar");
                }
                if (i < columnArr.length - 1) {
                    creteTableSQL.append(",");
                } else creteTableSQL.append(")");

            }
            System.out.println("建表语句:" + creteTableSQL);

            //预编译Sql
            preparedStatement = phoenixConn.prepareStatement(creteTableSQL.toString());

            //执行建表
            preparedStatement.execute();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("建表" + sinkTable + "失败");
        } finally {
            //归还连接
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (phoenixConn != null) {
                try {
                    phoenixConn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

        }
    }


}
